package day4

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{lit, window}
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}

object SparkStreaming {
  def main(args: Array[String]): Unit = {
    // 需要实现应用对于所收到数据的及时反馈
    // Spark Streaming专门针对流处理
    // Spark Streaming流处理的API跟批处理非常相似

    // DStream --> Structured Streaming：无边界表
    System.setProperty("hadoop.home.dir", "D:\\devtools\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)

    val session = SparkSession.builder()
      .appName("Spark Streaming")
      .master("local[*]")
      .getOrCreate()

    /**
     * <h1>数据处理的步骤</h1>
     *
     * 1 定义数据的来源
     *
     * 2 定义处理的方式
     *
     * 3 定义处理的输出
     *
     * 4 启动计算
     *
     */

    //1 定义数据的来源
    //需要虚拟机向端口输出数据，在Linux当中安装一个软件包：nmap-ncat
    //    val line = session.readStream.format("socket")
    //      .option("host","192.168.3.128")
    //      .option("port","9999")
    //      .load()

    //定义数据的来源为文件
    // 1 定义文件的格式
    val fileSchema = new StructType()
      .add("key",StringType)
      .add("value",StringType)

    val line = session.readStream.format("text")
      .schema(fileSchema)
      // spark streaming需要使用目录
      .load("D:\\data\\files\\wordCount")


    //2 定义处理方式
    import session.implicits._
    val get = line.select("value").as[String].flatMap(_.split(" "))
    //    val get = line.select("value").as[String]
    //      .flatMap(_.split(" "))
    //      .map((_,1))
    //      .groupByKey(_._1).count()
    /**
     * 每行独立完成的事情，比如：map、filter、select算子-->在处理新数据的时候不会去使用以前的数据，称为无状态操作
     *
     * 需要维护结果的状态进行跨多行处理，比如：count、分组、连接算子-->在处理新数据的时候会读取以前的数据，称之为有状态操作
     *
     * 流处理中独特的操作：窗口、触发器
     */

    //3 定义输出
    val put = get.writeStream
      .format("console")
      .outputMode("update")
      .trigger(Trigger.ProcessingTime("3 seconds"))

    /**
     * outputMode有三种选项：
     * complete: 每次输出都会将所有数据完整的输出；只支持输出内容比输入内容少，才能够实现将数据保存在内存中
     * append：只会输出新内容，以前内容不会再输出
     * update：只会输出上次处理结果变更之后的数据，绝大部分的操作都支持
     */

    /**
     * trigger：触发器
     * 固定内容：ProcessingTime指定间隔时间
     * Once：使用的是Once触发器，在接收到数据之后只处理一次，可以在外部再次定义Once触发器，实现触发自定义
     * Continuous：使用Continuous触发器连续触发
     */

    //4 执行
    put.start().awaitTermination()

  }
}